package cn.doitedu.api;

import beans.UserAction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/24
 * @Desc: 学大数据，上多易教育
 *
 *   接收行为数据
 *       {"session_id":"s01","uid":1,"timestamp":12309341,"event_id":"APP_LAUNCH"}
 *       {"session_id":"s01","uid":1,"timestamp":12309343,"event_id":"B"}
 *       {"session_id":"s01","uid":1,"timestamp":12309345,"event_id":"A"}
 *       {"session_id":"s01","uid":1,"timestamp":12309346,"event_id":"C"}
 *
 *     统计用户的会话时长
 *
 **/
public class _13_ProcessFunction_Api {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> stream = env.socketTextStream("doitedu", 8899);

        SingleOutputStreamOperator<UserAction> beans = stream.map(json -> JSON.parseObject(json, UserAction.class));

        beans.keyBy(UserAction::getSession_id)
                .process(new KeyedProcessFunction<String, UserAction, String>() {

                    ValueState<UserAction> firstState;
                    ValueState<UserAction> lastState;
                    ValueState<Long> timerState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        this.firstState = getRuntimeContext().getState(new ValueStateDescriptor<UserAction>("first", UserAction.class));
                        this.lastState = getRuntimeContext().getState(new ValueStateDescriptor<UserAction>("last", UserAction.class));
                        this.timerState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer", Long.class));
                    }

                    @Override
                    public void processElement(UserAction userAction, KeyedProcessFunction<String, UserAction, String>.Context ctx, Collector<String> out) throws Exception {

                        // 先取消之前注册的定时器
                        if(timerState.value() != null ) {
                            ctx.timerService().deleteProcessingTimeTimer(timerState.value());
                        }


                        if(firstState.value() == null ){
                            firstState.update(userAction);
                        }

                        lastState.update(userAction);

                        // 判断本次到达的事件，是否是 app_close 事件，如果是，说明会话已结束
                        if(userAction.getEvent_id().equals("APP_CLOSE")){
                            long duration = lastState.value().getTimestamp() - firstState.value().getTimestamp();
                            JSONObject res = new JSONObject();
                            res.put("uid",userAction.getUid());
                            res.put("session_id",ctx.getCurrentKey());  // 从此刻的ctx中就能取到当前数据所属的key
                            res.put("duration",duration);

                            out.collect(res.toJSONString());

                        }else{
                            // 注册5分钟之后的定时器
                            TimerService timerService = ctx.timerService();

                            long timerTime = System.currentTimeMillis() + 1*60*1000;
                            timerService.registerProcessingTimeTimer(timerTime);
                            timerState.update(timerTime);

                        }

                    }

                    @Override
                    public void close() throws Exception {

                    }

                    /**
                     *  是在注册的定时器被触发时调用
                     */
                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, UserAction, String>.OnTimerContext ctx, Collector<String> out) throws Exception {

                        System.out.println("定时器触发了");
                        long duration = lastState.value().getTimestamp() - firstState.value().getTimestamp();

                        JSONObject res = new JSONObject();
                        res.put("uid",firstState.value().getUid());
                        res.put("session_id",ctx.getCurrentKey());  // 从此刻的ctx中就能取到当前数据所属的key
                        res.put("duration",duration);

                        out.collect(res.toJSONString());
                    }
                }).print();


        env.execute();

    }

}
